package com.joe.xl.common.util.es;

import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.sum.Sum;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class ElasticClientPool {
    private final static Logger logger = LoggerFactory.getLogger(ElasticClientPool.class);

    private Settings setting;

    private Map<String, RestHighLevelClient> clientMap = new ConcurrentHashMap<String, RestHighLevelClient>();

    private Map<String, Integer> ips = new HashMap<>(); // hostname port

    private String clusterName = "test";

    private ElasticClientPool() {
        init();
    }

    /**
     * 初始化默认的client
     */
    public void init() {
        /*Properties prop = ReadConfig.loadPropertiesFromFile("/elastic.properties");
        String esHosts = prop.getProperty("elastic_topic.hosts");
        clusterName = prop.getProperty("cluster.name");

        if(StringUtils.isEmpty(esHosts)){
            prop = ReadConfig.load("/elastic.properties");
            esHosts = prop.getProperty("elastic_topic.hosts");
            clusterName = prop.getProperty("cluster.name");
        }*/
        String esHosts = "";

        System.out.println("connection es nodes : " + esHosts);

        initIps(esHosts);

        setting = Settings.builder()
                .put("cluster.name",clusterName)
                .build();
        addClient(setting, getAllAddress(ips));
    }

    public void initIps(String esHosts){
        String[] hosts = esHosts.split(",");
        for (String host : hosts) {
            String[] hostArr = host.split(":");
            ips.put(hostArr[0], Integer.parseInt(hostArr[1]));
        }
    }

    public void addClient(Settings setting, List<HttpHost> httpHosts) {
        try {
            RestClientBuilder restClientBuilder = RestClient.builder(httpHosts.toArray(new HttpHost[httpHosts.size()]));

            RestHighLevelClient client = new RestHighLevelClient(restClientBuilder);

            clientMap.put(setting.get("cluster.name"), client);

        }catch (Exception e){
            logger.error("elastic topic init error ... " + e.getMessage());
        }
    }

    /**
     * 获得所有的地址端口
     *
     * @return
     */
    public List<HttpHost> getAllAddress(Map<String, Integer> ips) {
        List<HttpHost> httpHosts = new ArrayList<>();
        for (String ip : ips.keySet()) {
            try {
                httpHosts.add(new HttpHost(ip, ips.get(ip), "http"));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return httpHosts;
    }

    public RestHighLevelClient getClient() {
        return getClient(clusterName);
    }

    public RestHighLevelClient getClient(String clusterName) {
        if (clientMap.get(clusterName) == null) {
            init();
        }
        return clientMap.get(clusterName);
    }

    public static final ElasticClientPool getInstance() {
        return SingleClient.INSTANCE;
    }

    private static class SingleClient {
        private static final ElasticClientPool INSTANCE = new ElasticClientPool();
    }
    
    public static void main(String[] args) {
//        ElasticClientPool.getInstance().getClient();
        logger.info("test slfj log...");
    }


    /***
     *
     * @throws Exception
     */
    public void demoEsQuery() throws Exception {

        BoolQueryBuilder qb = QueryBuilders.boolQuery();
        qb.must(QueryBuilders.termQuery("name", 1));
        qb.must(QueryBuilders.termQuery("type", 1));
        qb.must(QueryBuilders.rangeQuery("etlupdatetime.keyword").gt("2010"));

        TermsAggregationBuilder aggregation = AggregationBuilders.terms("by_date").field("Month.keyword").order(BucketOrder.key(true)).size(999)
                .subAggregation(AggregationBuilders.sum("sum_saleAmt").field("SaleAmt"))
                .subAggregation(AggregationBuilders.sum("sum_cost").field("Cost"))
                .subAggregation(AggregationBuilders.sum("sum_profit").field("Profit"));

                                              AggregationBuilders.terms("by_CalendarQuarter").field("Datekey.keyword").size(999)
                .subAggregation(AggregationBuilders.terms("by_cardType").field("CardType.keyword")
                        .subAggregation(AggregationBuilders.sum("sum_saleAmt").field("SaleAmt"))
                        .subAggregation(AggregationBuilders.cardinality("by_AchieveSysno").field("AchieveSysno")) // distinct
                ).order(BucketOrder.key(true));

        SearchSourceBuilder ssb = new SearchSourceBuilder();
        ssb.query(qb);
        ssb.aggregation(aggregation);
        ssb.sort("lastorderdate.keyword", SortOrder.DESC);
        ssb.from(1);
        ssb.size(20);

        SearchRequest searchRequest = new SearchRequest("index_es");
        searchRequest.source(ssb);

        RestHighLevelClient client   = ElasticClientPool.getInstance().getClient();
        SearchResponse      response = client.search(searchRequest, RequestOptions.DEFAULT);

        // 查询数据
        SearchHits          hits     = response.getHits();
        if (hits.getTotalHits() > 0) {
            for (SearchHit hit : hits) {
                Map<String, Object> map           = hit.getSourceAsMap();//可以转化成map再对操作数据
                int                 name   = Integer.parseInt(String.valueOf(map.getOrDefault("name", 0)));
                int                 cnt        = Integer.parseInt(String.valueOf(map.getOrDefault("cnt", 0)));
                double              pwd   = new BigDecimal(String.valueOf(map.getOrDefault("pwd", "0.00")))
                        .setScale(2, BigDecimal.ROUND_DOWN)
                        .doubleValue();
            }
        }

        // 聚合数据
        Aggregations       aggregations = response.getAggregations();
        Terms              terms        = aggregations.get("by_date");
        List<Terms.Bucket> list         = (List<Terms.Bucket>) terms.getBuckets();
        if (Objects.nonNull(list) && !list.isEmpty()) {
            for (int i = 0; i < list.size(); i++) {
                String       date             = list.get(i).getKeyAsString(); // 2019_02
                Aggregations aggregations_sum = list.get(i).getAggregations();

                Sum sum_saleAmt = aggregations_sum.get("sum_saleAmt");
                Sum sum_cost    = aggregations_sum.get("sum_cost");
                Sum sum_profit  = aggregations_sum.get("sum_profit");

                Map<String, Object> map = new HashMap<>();
                map.put("date", date);

                BigDecimal bg_saleamt = new BigDecimal(sum_saleAmt.getValueAsString());
                double     b1         = bg_saleamt.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();

                BigDecimal bg_cost = new BigDecimal(sum_cost.getValueAsString());
                double     b2      = bg_cost.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();

                BigDecimal bg_profit = new BigDecimal(sum_profit.getValueAsString());
                double     b3        = bg_profit.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();

                map.put("saleAmt", b1);
                map.put("cost", b2);
                map.put("profit", b3);

            }
        }

    }
}
